package com.zenitera.bigdata.doris;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink_SQL_doris_read {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql("create table flink_0520(" +
                " siteid int, " +
                " citycode SMALLINT, " +
                " username string, " +
                " pv bigint " +
                ")with(" +
                "  'connector' = 'doris', " +
                "  'fenodes' = 'hdt-dmcp-ops01:8130', " +
                "  'table.identifier' = 'test_db.flinktest', " +
                "  'username' = 'root', " +
                "  'password' = '123456' " +
                ")");

        tEnv.sqlQuery("select * from flink_0520").execute().print();

    }
}

/*
/*
   控制台输出信息：
+----+-------------+----------+---------------+---------+
| op |      siteid | citycode |      username |      pv |
+----+-------------+----------+---------------+---------+
| +I |           1 |        1 |           aaa |       1 |
| +I |           3 |        3 |           ccc |       3 |
| +I |           2 |        2 |           bbb |       2 |
| +I |           4 |        4 |      wangting |       4 |
+----+-------------+----------+---------------+---------+
4 rows in set
*/